package com.arch.bigdata.hbase;

import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.hbase.HBaseConfiguration;
import org.apache.hadoop.hbase.TableName;
import org.apache.hadoop.hbase.client.BufferedMutator;
import org.apache.hadoop.hbase.client.Connection;
import org.apache.hadoop.hbase.client.Get;
import org.apache.hadoop.hbase.client.Put;
import org.apache.hadoop.hbase.client.Result;
import org.apache.hadoop.hbase.client.Table;
import org.apache.hadoop.hbase.spark.JavaHBaseContext;
import org.apache.hadoop.hbase.util.Bytes;
import org.apache.spark.SparkConf;
import org.apache.spark.api.java.JavaRDD;
import org.apache.spark.api.java.JavaSparkContext;
import org.apache.spark.api.java.function.VoidFunction;
import scala.Tuple2;

import java.util.ArrayList;
import java.util.Iterator;
import java.util.List;

/**
 * @author pizhihui
 * @date 2020-08-25
 */
public class HbaseBasicSparkDemo {


    public static void main(String[] args) {


        SparkConf sparkConf = new SparkConf();

        JavaSparkContext jsc = new JavaSparkContext(sparkConf);

        try {
            List<byte[]> list = new ArrayList<>();
            list.add(Bytes.toBytes("1"));
            list.add(Bytes.toBytes("5"));

            JavaRDD<byte[]> rdd = jsc.parallelize(list);
            Configuration conf = HBaseConfiguration.create();

            JavaHBaseContext hbaseContext = new JavaHBaseContext(jsc, conf);

            String tableName = "";

//            hbaseContext.foreachPartition(rdd,
//                    new VoidFunction<Tuple2<Iterator<byte[]>, Connection>>() {
//                        @Override
//                        public void call(Tuple2<Iterator<byte[]>, Connection> t)
//                                throws Exception {
//                            Table table = t._2().getTable(TableName.valueOf(tableName));
//                            BufferedMutator mutator = t._2().getBufferedMutator(TableName.valueOf(tableName));
//                            while (t._1().hasNext()) {
//                                byte[] b = t._1().next();
//                                Result r = table.get(new Get(b));
//                                if (r.getExists()) {
//                                    mutator.mutate(new Put(b));
//                                }
//                            }
//
//                            mutator.flush();
//                            mutator.close();
//                            table.close();
//                        }
//                    });
        } finally {
            jsc.stop();
        }


    }


}
